package com.tcm.smarthealth.service.impl;

import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.tcm.smarthealth.common.exception.BusinessException;
import com.tcm.smarthealth.dto.QuestionCreateRequest;
import com.tcm.smarthealth.dto.QuestionResponse;
import com.tcm.smarthealth.dto.QuestionStreamChunk;
import com.tcm.smarthealth.entity.Question;
import com.tcm.smarthealth.enums.QuestionType;
import com.tcm.smarthealth.mapper.QuestionMapper;
import com.tcm.smarthealth.service.QuestionService;
import com.tcm.smarthealth.service.ai.AiClient;
import com.tcm.smarthealth.service.ai.AiClientException;
import com.tcm.smarthealth.service.ai.AiInsight;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.FluxSink;
import reactor.core.scheduler.Schedulers;
import reactor.core.scheduler.Scheduler;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.List;
import java.util.stream.Collectors;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicBoolean;

@Service
public class QuestionServiceImpl extends ServiceImpl<QuestionMapper, Question> implements QuestionService {

    private static final int DEFAULT_PAGE_SIZE = 10;
    private static final Logger log = LoggerFactory.getLogger(QuestionServiceImpl.class);

    private final AiClient aiClient;

    public QuestionServiceImpl(AiClient aiClient) {
        this.aiClient = aiClient;
    }

    @Override
    public QuestionResponse createQuestion(QuestionCreateRequest request, Long userId) {
        QuestionType type = QuestionType.fromCode(request.getType());
        String normalizedContent = request.getContent().trim();

        AiInsight insight = resolveQuestionInsight(type, normalizedContent);
        Question question = persistQuestion(userId, type, normalizedContent, insight);
        return toResponse(question);
    }

    @Override
    public List<QuestionResponse> listQuestions(Long userId, QuestionType type, int page, int size) {
        int pageNo = Math.max(page, 1);
        int pageSize = size > 0 ? size : DEFAULT_PAGE_SIZE;

        Page<Question> resultPage = lambdaQuery()
                .eq(Question::getUserId, userId)
                .eq(type != null, Question::getQaType, type != null ? type.name() : null)
                .orderByDesc(Question::getCreatedAt)
                .page(new Page<>(pageNo, pageSize));

        return resultPage.getRecords().stream()
                .map(this::toResponse)
                .collect(Collectors.toList());
    }

    @Override
    public Flux<QuestionStreamChunk> streamQuestion(QuestionCreateRequest request, Long userId) {
        QuestionType type = QuestionType.fromCode(request.getType());
        String normalizedContent = request.getContent().trim();

        return Flux.create(emitter -> {
            FluxSink<QuestionStreamChunk> sink = emitter;
            Scheduler emitterScheduler = Schedulers.newSingle("question-stream-emitter-" + System.nanoTime());

            AtomicReference<Disposable> finalTaskRef = new AtomicReference<>();
            AtomicReference<Disposable> streamRef = new AtomicReference<>();
            AtomicBoolean finished = new AtomicBoolean(false);
            AtomicBoolean cleaned = new AtomicBoolean(false);

            Runnable cleanup = () -> {
                if (cleaned.compareAndSet(false, true)) {
                    Disposable finalTask = finalTaskRef.getAndSet(null);
                    if (finalTask != null && !finalTask.isDisposed()) {
                        finalTask.dispose();
                    }
                    Disposable streamDisposable = streamRef.getAndSet(null);
                    if (streamDisposable != null && !streamDisposable.isDisposed()) {
                        streamDisposable.dispose();
                    }
                    emitterScheduler.dispose();
                }
            };

            Flux<String> source;
            try {
                source = aiClient.streamQuestionAnswer(normalizedContent, type);
            } catch (AiClientException ex) {
                log.error("AI streaming unavailable: {}", ex.getMessage(), ex);
                sink.error(new BusinessException(HttpStatus.BAD_GATEWAY, "AI 服务暂不可用，请稍后重试"));
                cleanup.run();
                return;
            }

            Disposable streamDisposable = source.subscribe(
                    delta -> emitterScheduler.schedule(() -> {
                        if (finished.get() || sink.isCancelled()) {
                            return;
                        }
                        QuestionStreamChunk chunk = new QuestionStreamChunk();
                        chunk.setContent(delta);
                        chunk.setFinished(false);
                        sink.next(chunk);
                    }),
                    error -> emitterScheduler.schedule(() -> {
                        if (finished.compareAndSet(false, true)) {
                            log.error("AI 流式输出失败", error);
                            if (!sink.isCancelled()) {
                                sink.error(new BusinessException(HttpStatus.BAD_GATEWAY, "AI 流式输出失败，请稍后重试"));
                            }
                            cleanup.run();
                        }
                    }),
                    () -> emitterScheduler.schedule(() -> {
                        if (finished.get() || sink.isCancelled()) {
                            cleanup.run();
                            return;
                        }
                        Disposable finalTask = Mono.fromCallable(() -> {
                                    AiInsight finalInsight = resolveQuestionInsight(type, normalizedContent);
                                    Question question = persistQuestion(userId, type, normalizedContent, finalInsight);
                                    QuestionResponse record = toResponse(question);
                                    return Map.entry(finalInsight, record);
                                })
                                .subscribeOn(Schedulers.boundedElastic())
                                .subscribe(result -> emitterScheduler.schedule(() -> {
                                            if (finished.compareAndSet(false, true) && !sink.isCancelled()) {
                                                AiInsight finalInsight = result.getKey();
                                                QuestionResponse record = result.getValue();

                                                QuestionStreamChunk completed = new QuestionStreamChunk();
                                                completed.setFinished(true);
                                                completed.setContent(record.getAnswer());
                                                completed.setConfidence(record.getConfidence());
                                                completed.setRecord(record);
                                                completed.setSuggestions(finalInsight.getSuggestions());
                                                sink.next(completed);
                                                sink.complete();
                                            }
                                            cleanup.run();
                                        }),
                                        ex -> emitterScheduler.schedule(() -> {
                                            if (finished.compareAndSet(false, true) && !sink.isCancelled()) {
                                                if (ex instanceof BusinessException) {
                                                    sink.error(ex);
                                                } else {
                                                    log.error("保存 AI 结果失败", ex);
                                                    sink.error(new BusinessException(HttpStatus.INTERNAL_SERVER_ERROR, "保存 AI 结果失败，请稍后重试"));
                                                }
                                            }
                                            cleanup.run();
                                        }));
                        finalTaskRef.set(finalTask);
                    })
            );

            streamRef.set(streamDisposable);

            sink.onDispose(() -> cleanup.run());
        }, FluxSink.OverflowStrategy.BUFFER);
    }

    private QuestionResponse toResponse(Question question) {
        QuestionResponse response = new QuestionResponse();
        response.setId(question.getId());
        response.setContent(question.getQuestion());
        response.setAnswer(question.getAnswer());
        response.setType(question.getQaType());
        response.setConfidence(question.getConfidence());
        response.setCreatedAt(question.getCreatedAt());
        return response;
    }

    private AiInsight resolveQuestionInsight(QuestionType type, String content) {
        try {
            AiInsight insight = aiClient.generateQuestionInsight(content, type);
            log.info("AI insight generated successfully for type={}, answer length={}", type,
                    insight.getAnswer() != null ? insight.getAnswer().length() : 0);
            return insight;
        } catch (AiClientException ex) {
            log.error("AI insight generation failed: {}", ex.getMessage(), ex);
            throw new BusinessException(HttpStatus.BAD_GATEWAY, "AI 服务暂不可用，请稍后重试");
        }
    }

    private Question persistQuestion(Long userId, QuestionType type, String questionContent, AiInsight insight) {
        Question question = new Question();
        question.setUserId(userId);
        question.setQuestion(questionContent);
        question.setQaType(type.name());
        question.setAnswer(composeAnswer(insight));
        question.setConfidence(normalizeConfidence(insight.getConfidence(), type));
        question.setCreatedAt(LocalDateTime.now());
        save(question);
        return question;
    }

    private String composeAnswer(AiInsight insight) {
        StringBuilder sb = new StringBuilder();
        if (StringUtils.hasText(insight.getAnswer())) {
            sb.append(insight.getAnswer().trim());
        } else if (StringUtils.hasText(insight.getDiagnosis())) {
            sb.append(insight.getDiagnosis().trim());
        }
        if (!CollectionUtils.isEmpty(insight.getSuggestions())) {
            if (sb.length() > 0) {
                sb.append("\n\n");
            }
            sb.append("建议：");
            for (String suggestion : insight.getSuggestions()) {
                sb.append("\n· ").append(suggestion);
            }
        }
        return sb.length() > 0 ? sb.toString() : "请在专业医生指导下进行进一步诊断。";
    }

    private BigDecimal normalizeConfidence(BigDecimal value, QuestionType type) {
        if (value != null && value.compareTo(BigDecimal.ZERO) > 0) {
            return value;
        }
        return BigDecimal.valueOf(0.9);
    }
}

